package com.gjy.rabbitmq.resolve;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Envelope;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * 幂等消息处理器
 * 使用Redis实现分布式环境下的消息幂等性保证
 * 防止消息在分布式系统中重复消费
 *
 * @author gjy
 * @version 1.0
 * @since 2025-10-13 14:41:58
 */
public class IdempotentMessageProcessor implements MessageProcessor {
    private static final Logger logger = LoggerFactory.getLogger(IdempotentMessageProcessor.class);

    /**
     * Redis工具类
     */
    private final RedisUtil redisUtil;

    /**
     * 业务服务
     */
    private final BusinessService businessService;

    /**
     * Redis键前缀
     */
    private static final String REDIS_KEY_PREFIX = "processed_message:";

    /**
     * 构造幂等消息处理器
     *
     * @param redisUtil       Redis工具类
     * @param businessService 业务服务
     */
    public IdempotentMessageProcessor(RedisUtil redisUtil, BusinessService businessService) {
        this.redisUtil = redisUtil;
        this.businessService = businessService;
        logger.info("幂等消息处理器初始化完成");
    }

    /**
     * 处理消息（幂等性保证）
     *
     * @param message    消息内容
     * @param properties 消息属性
     * @param envelope   交付信息
     * @return 处理是否成功
     * @throws Exception 处理异常
     */
    @Override
    public boolean process(String message, AMQP.BasicProperties properties, Envelope envelope)
            throws Exception {
        String messageId = properties.getMessageId();

        // 构造Redis键
        String redisKey = REDIS_KEY_PREFIX + messageId;

        // 使用Redis实现分布式幂等性
        // 设置成功表示首次处理，设置失败表示已处理过
        boolean isFirstProcess = redisUtil.setIfAbsent(redisKey, "1", 24, TimeUnit.HOURS);

        if (!isFirstProcess) {
            // 消息已处理过，直接返回成功
            logger.info("消息已通过Redis标记为已处理: messageId={}", messageId);
            return true;
        }

        try {
            // 执行业务逻辑
            businessService.processBusinessLogic(message, properties);

            // 业务处理成功，返回true
            logger.info("业务逻辑处理成功: messageId={}", messageId);
            return true;

        } catch (Exception e) {
            // 业务处理失败，删除Redis标记以便重试
            logger.warn("业务逻辑处理失败，删除Redis标记以便重试: messageId={}", messageId, e);
            redisUtil.delete(redisKey);
            throw e;
        }
    }

}
